package dim

import java.nio.charset.Charset
import java.util

import com.alibaba.fastjson.JSON
import modes.{AreaInfo, DimAreaInfo, OrderInfo, TableObject}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.hadoop.conf
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Put, Table}
import org.apache.hadoop.hbase.util.Bytes

class AreaHBaseSink extends RichSinkFunction[String]{
  var connection: Connection = _
  var hTable: Table = _
  val tName = "dim_lagou_area"
  val family = "f1"

  override def open(parameters: Configuration): Unit = {
    val configuration: conf.Configuration = HBaseConfiguration.create()
    configuration.set("hbase.zookeeper.quorum","linux100")
    configuration.set("hbase.zookeeper.property.clientPort", "2181")
    connection = ConnectionFactory.createConnection(configuration)
    hTable = connection.getTable(TableName.valueOf(tName))
  }

  override def invoke(value: String, context: SinkFunction.Context[_]): Unit = {
    val items: Array[String] = value.split(",")
    insertIntoDimArea(hTable, DimAreaInfo(items(0).toInt, items(1), items(2).toInt, items(3), items(4).toInt, items(5)))
  }

  override def close(): Unit = {
    if (hTable != null) {
      hTable.close()
    }

    if (connection != null) {
      connection.close()
    }
  }

  def insertIntoDimArea(hTable: Table, dimAreaInfo: DimAreaInfo): Unit = {
    val put = new Put(Bytes.toBytes(dimAreaInfo.areaId))
    put.addColumn(family.getBytes(Charset.forName("UTF-8")), "areaName".getBytes(Charset.forName("UTF-8")) ,dimAreaInfo.areaName.getBytes(Charset.forName("UTF-8")))
    put.addColumn(family.getBytes(Charset.forName("UTF-8")), "cityId".getBytes(Charset.forName("UTF-8")) , Bytes.toBytes(dimAreaInfo.cityId))
    put.addColumn(family.getBytes(Charset.forName("UTF-8")), "cityName".getBytes(Charset.forName("UTF-8")) ,dimAreaInfo.cityName.getBytes(Charset.forName("UTF-8")))
    put.addColumn(family.getBytes(Charset.forName("UTF-8")), "provinceId".getBytes(Charset.forName("UTF-8")) , Bytes.toBytes(dimAreaInfo.provinceId))
    put.addColumn(family.getBytes(Charset.forName("UTF-8")), "provinceName".getBytes(Charset.forName("UTF-8")) ,dimAreaInfo.provinceName.getBytes(Charset.forName("UTF-8")))
    hTable.put(put)
  }
}
